package com.adingxiong.cft.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * @ClassName CronTimer
 * @Description  定时启动和停止 消息监听器
 * @Author xiongchao
 * @Date 2021/3/25 14:43
 **/
@EnableScheduling
@Component
@Slf4j
public class CronTimer {

    private final String CONSUMERID ="timingConsumer";

    @Autowired
    private KafkaListenerEndpointRegistry registry;

    @Autowired
    private ConsumerFactory factory;

    @Bean
    public ConcurrentKafkaListenerContainerFactory delayContainerFactory() {
        ConcurrentKafkaListenerContainerFactory containerFactory = new ConcurrentKafkaListenerContainerFactory();
        containerFactory.setConsumerFactory(factory);
        //机制kafka监听器自动启动
        containerFactory.setAutoStartup(false);
        return containerFactory;
    }

    @KafkaListener(id = CONSUMERID ,topics ="{topic1,topic2,test",containerFactory = "delayContainerFactory")
    public void onMessage(ConsumerRecord<?,?> record) {
        log.info("topic:{},value:{}.time:{}",record.topic(),record.value(),record.timestamp());
    }


    @Scheduled(cron = "0 30 8 * * ? ")
    public void startListener() {
        log.info("开始启动监听器.当前时间为:{}",new Date().toString());
        if(!registry.getListenerContainer(CONSUMERID).isRunning()){
            registry.getListenerContainer(CONSUMERID).start();
        }
    }

    @Scheduled(cron = "0 30 18 * * ? ")
    public void endListener() {
        log.info("关闭监听器.当前时间为:{}",new Date().toString());
        registry.getListenerContainer(CONSUMERID).pause();
    }

}
